package com.yusito.hadoop.hdfs;

import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;

public class HDFSClient {

    /**
     * 日志对象
     */
    protected static final Log log = LogFactory.getLog(HDFSClient.class);

    private FileSystem fileSystem;

    /**
     * 无参构造，初始化文件系统
     */
    public HDFSClient() {
        fileSystem = HDFSServer.getFileSystem();
    }

    /**
     * 关闭文件系统
     * @throws IOException
     */
    private void close() throws IOException {
        fileSystem.close();
    }

    /**
     * 文件上传
     * @param src 源文件路径
     * @param dst 上传路径
     * @throws IOException
     */
    private void upload(String src,String dst) throws IOException {
        fileSystem.copyFromLocalFile(new Path(src),new Path(dst));
    }

    /**
     * 文件上传
     * @param src 源文件路径
     * @param dst 上传路径
     * @throws IOException
     */
    private void uploadByStream(String src,String dst) throws IOException {
        final FileInputStream inputStream = new FileInputStream(src);
        final FSDataOutputStream outputStream = this.fileSystem.create(new Path(dst), true);
        IOUtils.copy(inputStream, outputStream);
    }

    /**
     * 文件下载
     * @param src 源文件路径
     * @param dst 下载路径
     * @throws IOException
     */
    private void download(String src,String dst) throws IOException {
        fileSystem.copyToLocalFile(new Path(src),new Path(dst));
    }

    /**
     * 文件下载
     * @param src 源文件路径
     * @param dst 下载路径
     * @throws IOException
     */
    private void downloadByStream(String src,String dst) throws IOException {
        final FSDataInputStream inputStream = this.fileSystem.open(new Path(src));
        final FileOutputStream outputStream = new FileOutputStream(dst);
        IOUtils.copy(inputStream, outputStream);
    }

    /**
     * 文件下载
     * @param src 源文件路径
     * @param dst 下载路径
     * @throws IOException
     */
    private void downloadByStreamAccess(String src,String dst,long offset) throws IOException {
        final FSDataInputStream inputStream = this.fileSystem.open(new Path(src));
        inputStream.seek(offset);
        final FileOutputStream outputStream = new FileOutputStream(dst);
        IOUtils.copy(inputStream, outputStream);
    }

    /**
     * 创建目录
     * @param dir
     */
    public boolean mkdirs(String dir) throws IOException {
        Path path = new Path(dir);
        if (fileSystem.exists(path)) {
            return true;
        }
        return fileSystem.mkdirs(path);
    }

    /**
     * 文件重命名
     * @param src 源文件名 /oldname
     * @param dst 新文件名 /newname
     * @return
     * @throws IOException
     */
    public boolean rename(String src, String dst) throws IOException {
        Path srcPath = new Path(src);
        if (fileSystem.exists(srcPath)) {
            Path dstPath = new Path(dst);
            return fileSystem.rename(srcPath, dstPath);
        }
        return false;
    }

    /**
     * 获取文件
     * @param filePath
     * @return
     * @throws IOException
     */
    public InputStream getFile(String filePath) throws IOException {
        return fileSystem.open(new Path(filePath));
    }

    public static void main(String[] args) {
        try {
            HDFSClient client = new HDFSClient();
            //client.mkdirs("/test/01");
            client.close();
        }catch (Exception e){

        }

    }
}
